package com.ytktt.study.week12.task4.kafka.consumer;

import com.ytktt.study.week12.task4.Order;
import com.ytktt.study.week12.task4.api.ConsumerService;
import com.ytktt.study.week12.task4.redis.RedisOperator;
import com.ytktt.study.week12.task4.redis.constant.RedisConstant;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;

import java.util.Collections;

/**
 * kafka 生产者有业务逻辑实现
 *
 * @author ytk
 */
@Slf4j
@RequiredArgsConstructor
public class KafkaConsumerServiceImpl implements ConsumerService {

    private final RedisOperator redisOperator;

    @Override
    @KafkaListener(id = "${spring.application.name}",
            topicPartitions = @TopicPartition(topic = "${financial.mq.kafka.topic}",
            partitions = "${financial.mq.kafka.consumer.partition}"),
            errorHandler = "${financial.mq.kafka.consumer.errorHandler}")
    public void consumer(ConsumerRecord<String, Order> consumerRecord, Acknowledgment ack, Consumer consumer) {
        Order order = consumerRecord.value();
        String orderId = String.valueOf(order.getId());
        boolean lockResult = this.redisOperator.lockConsumerState(orderId);
        if (lockResult) {
            this.handleConsume(orderId);
        } else {
            String state = this.redisOperator.get(orderId);
            if (RedisConstant.CONSUMING.equals(state)) {
                log.info("order {} is consuming, try consume later, message info: {}", orderId, consumerRecord);
            } else if (RedisConstant.CONSUMED.equals(state)) {
                log.info("order {} has been consumed", orderId);
            } else {
                this.handleConsume(orderId);
            }
        }
        ack.acknowledge();

        if (consumerRecord.partition() == 0) {
            for (int i = 0; i < 3; i++) {
                org.apache.kafka.common.TopicPartition tp = new org.apache.kafka.common.TopicPartition(consumerRecord.topic(), consumerRecord.partition());
                consumer.seekToBeginning(Collections.singletonList(tp));
            }
        }
    }

    private boolean handleConsume(String orderId) {
        boolean consumeResult = false;
        try {
            log.info("order {} consumed", orderId);
            consumeResult = true;
        } catch (Exception e) {
            this.redisOperator.deleteConsumerState(orderId);
            throw e;
        }

        if (consumeResult) {
            this.redisOperator.markConsumed(orderId);
        } else {
            this.redisOperator.deleteConsumerState(orderId);
        }

        return consumeResult;
    }

}
